import org.apache.spark.sql.{SparkSession, DataFrame}

object SparkHDFSOperations {
  def main(args: Array[String]): Unit = {
    // 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("Spark HDFS CSV Operations")
      .getOrCreate()

    import spark.implicits._

    // HDFS路径配置（根据实际环境修改）
    val hdfsNamenode = "kkwang"
    val hdfsPort = "8020"
    val inputPath = s"hdfs://$hdfsNamenode:$hdfsPort/input/electronic_product_sales.csv"
    val outputPath = s"hdfs://$hdfsNamenode:$hdfsPort/output/sales_analysis_result"

    try {
      // 1. 从HDFS读取CSV文件
      val csvData: DataFrame = spark.read
        .option("header", "true") // 表示CSV文件包含表头
        .option("inferSchema", "true") // 自动推断列的数据类型
        .option("sep", ",") // 指定CSV分隔符
        .option("quote", "\"") // 指定引号字符
        .option("escape", "\"") // 指定转义字符
        .csv(inputPath)

      // 显示读取的数据信息
      println(s"成功读取CSV文件，共 ${csvData.count()} 行数据")
      println("数据 schema:")
      csvData.printSchema()
      println("前5行数据:")
      csvData.show(5, truncate = false)

      // 2. 执行简单的计算示例（按性别统计平均价格）
      val resultDF = csvData
        .groupBy("sex")
        .avg("price")
        .withColumnRenamed("avg(price)", "average_price")
        .orderBy("sex")

      println("计算结果:")
      resultDF.show()

      // 3. 将计算结果写入HDFS
      resultDF.write
        .option("header", "true") // 输出文件包含表头
        .mode("overwrite") // 覆盖已存在的输出路径
        .option("sep", ",") // 指定输出分隔符
        .csv(outputPath)

      println(s"计算结果已成功写入HDFS路径: $outputPath")

      // 验证输出结果
      val verifyDF = spark.read
        .option("header", "true")
        .csv(outputPath)

      println("验证输出结果:")
      verifyDF.show()

    } catch {
      case e: Exception =>
        println(s"操作过程中发生错误: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 关闭SparkSession
      spark.stop()
    }
  }
}
